package com.xiang.community.message;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xiang.community.bo.MessageBO;
import com.xiang.community.dto.UserFocusDTO;
import com.xiang.community.entity.Message;
import com.xiang.community.entity.User;
import com.xiang.community.enums.MessageStatusEnum;
import com.xiang.community.enums.UserFocusStatusEnum;
import com.xiang.community.service.DiscussPostService;
import com.xiang.community.service.MessageService;
import com.xiang.community.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.Map;
import java.util.UUID;

import static com.xiang.community.constant.KafkaConstant.*;
import static com.xiang.community.constant.LoginConstant.FOLLOWEE_COUNT;
import static com.xiang.community.constant.LoginConstant.FOLLOWER_COUNT;

/**
 * @author qixiangLi
 * @Date 2021/11/30
 */
@Component
@Slf4j
public class MessageConsumer {
    @Resource
    private DiscussPostService discussPostService;
    @Resource
    private UserService userService;
    @Resource
    private ObjectMapper mapper;
    @Resource
    private MessageService messageService;

    @KafkaListener(topics = {TOPIC_GOOD})
    public void handleGoodMessage(ConsumerRecord record) {
        try {
            MessageBO messageBO = mapper.readValue(record.value().toString(), MessageBO.class);
            // 更新数据库，后续使用MQ
            discussPostService.goodPost(Integer.parseInt(messageBO.getData().get("discussPostId").toString()));
            // 更新发帖人的点赞数
            User poster = userService.findUserById(messageBO.getToUserId());
            userService.updateField("good", poster.getGood() + 1, poster.getId());
        } catch (JsonProcessingException e) {
            log.error(e.toString());
        }
    }

    @KafkaListener(topics = {TOPIC_FOLLOW})
    public void handleFollowMessage(ConsumerRecord record) {
        try {
            MessageBO messageBO = mapper.readValue(record.value().toString(), MessageBO.class);
            User userOne = userService.findUserById(Integer.parseInt(messageBO.getData().get("userOneId").toString()));
            User userTwo = userService.findUserById(Integer.parseInt(messageBO.getData().get("userTwoId").toString()));
            UserFocusDTO userFocusDTO = userService.selectFocus(userOne.getId(), userTwo.getId());
            if (userFocusDTO == null) {
                userFocusDTO = UserFocusDTO.builder().userTwoId(userTwo.getId()).userOneId(userOne.getId())
                        .createTime(new Date()).build();
                userService.insertFocus(userFocusDTO);
            } else {
                userService.updateFocus(userFocusDTO.getId(), UserFocusStatusEnum.THREE.getStatus());
            }
            // 更新关注数据
            userService.updateField(FOLLOWER_COUNT, userOne.getFollowerCount() + 1, userOne.getId());
            userService.updateField(FOLLOWEE_COUNT, userTwo.getFolloweeCount() + 1, userTwo.getId());
        } catch (JsonProcessingException e) {
            log.error(e.toString());
        }
    }

    @KafkaListener(topics = {TOPIC_UNFOLLOW})
    public void handleUnfollowMessage(ConsumerRecord record) {
        try {
            MessageBO messageBO = mapper.readValue(record.value().toString(), MessageBO.class);
            User userOne = userService.findUserById(Integer.parseInt(messageBO.getData().get("userOneId").toString()));
            User userTwo = userService.findUserById(Integer.parseInt(messageBO.getData().get("userTwoId").toString()));
            // 更新关注数据
            userService.updateField(FOLLOWER_COUNT, userOne.getFollowerCount() - 1, userOne.getId());
            userService.updateField(FOLLOWEE_COUNT, userTwo.getFolloweeCount() - 1, userTwo.getId());
            // 更新关注关系
            UserFocusDTO userFocusDTO = userService.selectFocus(userOne.getId(), userTwo.getId());
            int status = userFocusDTO.getStatus();
            status = userOne.getId() == userFocusDTO.getUserOneId() ? status - 2 : status - 1;
            if (status <= 0) {
                // 两者没有关注关系
                userService.deleteFocus(userFocusDTO.getId());
            } else {
                // 对方关注自己
                userService.updateFocus(userFocusDTO.getId(), status);
            }
        } catch (JsonProcessingException e) {
            log.error(e.toString());
        }
    }

    @KafkaListener(topics = {TOPIC_COMMENT})
    public void handleCommentMessage(ConsumerRecord record) {
        try {
            MessageBO messageBO = mapper.readValue(record.value().toString(), MessageBO.class);
            Map<String, Object> map = messageBO.getData();
            StringBuilder sb = new StringBuilder();
            sb.append(messageBO.getFromUserName()).append(" 在你的帖子 ").append(map.get("discussPostTitle"))
                    .append("下说: ").append(map.get("commentContent"));
            Message message = Message.builder().content(sb.toString())
                    .status(MessageStatusEnum.UNREAD.getStatus())
                    .conversationId(UUID.randomUUID().toString())
                    .createTime(new Date())
                    .fromId(1).toId(messageBO.getToUserId()).build();
            messageService.addMessage(message);
        } catch (JsonProcessingException e) {
            log.error(e.toString());
        }
    }
}
